package com.lanmao.emc.strategy.impl;

import com.lanmao.emc.entity.FeIndexLogPO;
import com.lanmao.emc.entity.FeVersionPO;
import com.lanmao.emc.entity.VersionInfo;
import com.lanmao.emc.entity.VersionUsedInfo;
import com.lanmao.emc.enums.DBType;
import com.lanmao.emc.enums.UpgradeType;
import com.lanmao.emc.utils.EsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ResourceUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

import static com.lanmao.emc.constant.FieldConstant.TABLE_INDEX_LOG;
import static com.lanmao.emc.constant.FieldConstant.TABLE_VERSION;
import static com.lanmao.emc.utils.UpgradeUtil.getEsTable;
import static java.nio.charset.StandardCharsets.UTF_8;

/**
 * 描述：版本管理策略，灵活度最高，最复杂的模式
 *
 * @author lanmao
 * @since 2024年11月12日 20:35
 */
@Slf4j
@Service
public class MultiVersionStrategyImpl extends UpgradeStrategyImpl {
    @Value("${spring.application.name}")
    private String projectName;

    @Value("${es.upgrade.count:1}")
    private int saveCount;
    /**
     * 版本更新需要保存的记录
     */
    private final List<FeIndexLogPO> saveList = new ArrayList<>();

    /**
     * 当前版本更新后需要被修改状态的索引记录，使用状态置零
     */
    private final List<String> updateIndexList = new ArrayList<>();

    /**
     * 设置了备份数量后，超过备份数量的索引以及记录都会被删除
     */
    private final List<String> delRelIndexList = new ArrayList<>();

    @Override
    void handleFailedIndex() {
        log.info("start handle failed index");
        List<VersionInfo> list = factory.getDao().findAllLastErrorIndex();
        int successCount = 0;
        if (CollectionUtils.isEmpty(list)) {
            log.info("no failed index to be upgrade");
            return;
        }
        Map<String, List<VersionInfo>> versionMap = list.stream().collect(Collectors.groupingBy(VersionInfo::getVersion));
        List<FeVersionPO> versionList = factory.getDao().findAllVersion();
        Map<String, String> dirNameMap = versionList.stream().collect(Collectors.toMap(FeVersionPO::getVersion, FeVersionPO::getRemark));
        for (Map.Entry<String, List<VersionInfo>> entry : versionMap.entrySet()) {
            String version = entry.getKey();
            Set<String> errorIndexSet = entry.getValue().stream().map(VersionInfo::getIndexName).collect(Collectors.toSet());
            for (Resource resource : getUpgradeDir(dirNameMap.get(version))) {
                String indexName = Objects.requireNonNull(resource.getFilename()).split("\\.")[0];
                if (errorIndexSet.contains(indexName)) {
                    String newIndexName = indexName + "_" + version;
                    try (InputStream inputStream = resource.getInputStream()) {
                        String json = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
                        successCount += upgradeIndex(indexName, json, newIndexName);
                    } catch (Exception e) {
                        log.error("upgrade index {} error", indexName, e);
                    }
                }
            }
            doFinally();
        }
        log.info("upgrade failed index count :{} ", successCount);
    }

    @Override
    void doFinally() {
        //移除需要被删除的多余版本历史
        if (!CollectionUtils.isEmpty(delRelIndexList)) {
            factory.getDao().deleteByIndexRelName(delRelIndexList);
            delRelIndexList.clear();
        }
        //修改历史版本的使用状态为0
        if (!CollectionUtils.isEmpty(updateIndexList)) {
            factory.getDao().updateIsUsedByIndexName(updateIndexList);
            updateIndexList.clear();
        }
        //保存升级成功记录数据
        if (!CollectionUtils.isEmpty(saveList)) {
            factory.getDao().batchSave(saveList);
            saveList.clear();
        }
    }

    @Override
    public void rollback(String index, String version) {
        List<VersionUsedInfo> indexVersionList = factory.getDao().findAllVersionByIndexName(index);
        String rollbackIndexName = indexVersionList.stream().map(VersionUsedInfo::getIndexRelName)
                .filter(x -> x.split("_")[1].equals(version)).findAny().orElse("");
        String currIndexName = indexVersionList.stream().filter(x -> x.getIsUsed() == 1).map(VersionUsedInfo::getIndexRelName).findFirst().orElse("");
        if (StringUtils.hasText(rollbackIndexName) && StringUtils.hasText(currIndexName)) {
            try {
                List<String> indexAlias = EsUtils.getIndexAlias(currIndexName);
                EsUtils.moveAliasToNewIndex(currIndexName, rollbackIndexName, indexAlias.toArray(new String[0]));
            } catch (IOException e) {
                log.error("rollback index {} error", index, e);
                return;
            }
            factory.getDao().updateIsUsedByIndexAndVersion(index, version);
            log.info("index {} rollback success,current version is {}", index, version);
            return;
        }
        log.error("索引{}没有该版本信息：{}", index, version);
    }

    @Override
    int upgradeIndex(String indexName, String json, String newIndexName) {
        long l = System.currentTimeMillis();
        String version = newIndexName.substring(newIndexName.lastIndexOf("_") + 1);
        String oldIndexName = indexName;
        long dataCount = 0L;
        try {
            EsUtils.addIndex(newIndexName, json);
            log.debug("create index {} success", newIndexName);
            //已经存在正常名称索引了，使用纳入版本管理的新索引
            if (EsUtils.existsIndex(indexName)) {
                dataCount = EsUtils.reindexAndGetCount(indexName, newIndexName);
                log.debug("reindex index {} success,count:{}", newIndexName, dataCount);
                List<VersionUsedInfo> indexVersionList = factory.getDao().findAllVersionByIndexName(indexName);
                if (CollectionUtils.isEmpty(indexVersionList)) {
                    EsUtils.delIndex(indexName);
                    log.debug("index {} is deleted", indexName);
                    EsUtils.createAlias(newIndexName, indexName);
                    log.debug("move alias to new index {} success", indexName);
                } else {
                    VersionUsedInfo currVersionInfo = indexVersionList.stream().filter(x -> x.getIsUsed() == 1)
                            .findAny().orElse(null);
                    if (currVersionInfo == null) {
                        log.error("index {}'s is_used is error need to handle", indexName);
                        return 0;
                    }
                    oldIndexName = currVersionInfo.getIndexRelName();
                    if (currVersionInfo.getIndexRelName().equals(newIndexName)) {
                        log.debug("database exist upgrade index:{}", newIndexName);
                        return 1;
                    }
                    EsUtils.moveAliasToNewIndex(currVersionInfo.getIndexRelName(), newIndexName, indexName);
                    log.debug("move alias to new index {} success", newIndexName);
                    List<String> indexAlias = EsUtils.getIndexAlias(currVersionInfo.getIndexRelName());
                    if (!CollectionUtils.isEmpty(indexAlias)) {
                        EsUtils.removeAlias(currVersionInfo.getIndexRelName(), indexAlias.toArray(new String[0]));
                        log.debug("{} all alias remove success", currVersionInfo.getIndexRelName());
                    }
                    removeExtraVersion(indexVersionList);
                }
            } else {
                EsUtils.createAlias(newIndexName, indexName);
                log.debug("add index {} alias success", newIndexName);
            }
        } catch (Exception e) {
            int result = handleException(oldIndexName, indexName, newIndexName, e);
            if (result == 0) {
                saveToLog(indexName, version, l, newIndexName, 0, "f", dataCount, e.getMessage());
                return result;
            }
        }
        updateIndexList.add(indexName);
        saveToLog(indexName, version, l, newIndexName, 1, "t", dataCount, null);
        log.info("index {} upgrade success", newIndexName);
        return 1;
    }

    @Override
    public void initTable() {
        if (!factory.getDao().allTableExist(TABLE_VERSION, TABLE_INDEX_LOG)) {
            if (factory.getDao().getDbType() == DBType.ES) {
                Resource resource = new PathMatchingResourcePatternResolver()
                        .getResource(ResourceUtils.CLASSPATH_URL_PREFIX + factory.getDao().getDbType().getType() + "/" + "version.json");
                Resource resource1 = new PathMatchingResourcePatternResolver()
                        .getResource(ResourceUtils.CLASSPATH_URL_PREFIX + factory.getDao().getDbType().getType() + "/" + "indexLog.json");

                createSourceTable(resource, getEsTable(TABLE_VERSION));
                createSourceTable(resource1, getEsTable(TABLE_INDEX_LOG));
                return;
            }
            Resource resource = new PathMatchingResourcePatternResolver()
                    .getResource(ResourceUtils.CLASSPATH_URL_PREFIX + factory.getDao().getDbType() + "/" + "multiVersion.sql");
            List<String> sqlList = readSqlFile(resource);
            factory.getDao().batchupdate(sqlList);
        }

    }

    private static void createSourceTable(Resource resource, String tableName) {
        String json = "";
        try (InputStream inputStream = resource.getInputStream()) {
            json = IOUtils.toString(inputStream, UTF_8);
            EsUtils.delIndex(tableName);
        } catch (Exception ignored) {
        } finally {
            try {
                EsUtils.addIndex(tableName, json);
            } catch (Exception e) {
                log.error("create {} and {} error", tableName, tableName, e);
            }
        }
    }

    @Override
    public UpgradeType getUpgradeType() {
        return UpgradeType.MULTI_VERSION;
    }

    /**
     * 移除多余版本
     *
     * @param indexVersionList 目前保留版本数
     */
    private void removeExtraVersion(List<VersionUsedInfo> indexVersionList) {
        if (saveCount > 0 && indexVersionList.size() >= saveCount) {
            List<String> deleteList = indexVersionList.stream().skip(saveCount - 1).map(VersionUsedInfo::getIndexRelName).distinct().toList();
            for (String index : deleteList) {
                try {
                    EsUtils.delIndex(index);
                } catch (Exception e) {
                    log.error("remove {} history version error,", index, e);
                }
            }
            delRelIndexList.addAll(deleteList);
        }
    }

    /**
     * 保存索引升级记录
     *
     * @param indexName    索引名
     * @param version      版本
     * @param startTime    升级开始时间
     * @param relIndexName 真实的索引名
     * @param isUsed       使用状态
     * @param status       升级成功与否
     * @param dataCount    升级同步数据总量
     * @param error        错误信息
     */
    private void saveToLog(String indexName, String version, long startTime, String relIndexName, int isUsed, String status, long dataCount, String error) {
        FeIndexLogPO indexLog = new FeIndexLogPO();
        indexLog.setVersion(version);
        indexLog.setCount(dataCount);
        indexLog.setIndexName(indexName);
        indexLog.setIndexRelName(relIndexName);
        indexLog.setIsUsed(isUsed);
        indexLog.setSuccess(status);
        indexLog.setCostTime(System.currentTimeMillis() - startTime);
        indexLog.setErrorInfo(error);
        indexLog.setProjectName(projectName);
        indexLog.setCreateTime(LocalDateTime.now());
        saveList.add(indexLog);
    }

    @Override
    String getNewIndexName(String index) {
        return index;
    }
}